package com.xinqing.bigdata.test.datastream

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}

/**
  * @Author:CHQ
  * @Date:2020 /6/22 15:32
  * @Description
  */
class MyJdbcSinkFun extends RichSinkFunction[SensorReading] {

  var conn: Connection = _
  var insertStatment: PreparedStatement = _
  var updateStatment: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://10.201.7.174:3307/test", "PT_super_user","Hb86be03cc_H3,,2c8ff86#326c3")
    insertStatment = conn.prepareStatement("insert into temperatures(id,temperature) values(?,?)")
    updateStatment = conn.prepareStatement("update temperatures set temperature=? where id=?")
  }

  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    updateStatment.setDouble(1, value.temperature)
    updateStatment.setString(2, value.id)
    val updateCount = updateStatment.executeUpdate()
    if (updateCount == 0) {
      insertStatment.setString(1, value.id)
      insertStatment.setDouble(2, value.temperature)
      insertStatment.execute()
    }
  }

  override def close(): Unit = {
    insertStatment.close()
    updateStatment.close()
    conn.close()
  }

}
